package com.burges.net.dataStream.codeRuler.transform.partition

import org.apache.flink.api.common.functions.Partitioner
import org.apache.flink.api.java.utils.ParameterTool
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment, _}

/**
  * 物理分区操作
  */
object PartitioningTransformation {

  def main(args: Array[String]): Unit = {
    // Checking input parameters
    val params = ParameterTool.fromArgs(args)

    // set up the execution environment
    val env = StreamExecutionEnvironment.getExecutionEnvironment

    // make parameters available in the web interface
    env.getConfig.setGlobalJobParameters(params)

    val dataStream: DataStream[(String, Int)] = env.fromElements(("a", 3), ("d", 4), ("c", 2), ("c", 5), ("a", 5))

    // 随机分区 RandomPartitioning 随机的方式将数据分配在下游算子的每个分区汇总，分区相对均衡，但是较容易失去原来数据的分区结构
    val shuffleStream = dataStream.shuffle

    // 按比例分区 RoundrobinPartitioning 通过循环数据的方式对数据集的数据进行重分区
    //  能够尽可能保证每个分区的数据平衡，当数据集发生倾斜带回后，使用该策略是比较有效的优化方法
    val rebalanceStream = dataStream.rebalance

    // RescalingPartitioning 通过循环的方式进行数据充平衡的策略
    // 与上面不同的是，上面是对全局性的通过网络介质传输到其他的节点完成数据的重新平衡
    // 而RescalingPartitioning仅会对上下游集成的算子数据进行重平衡，具体的分区住哟啊根据上下游算子的并行度决定
    val rescaleStream = dataStream.rescale

    // 广播操作broadcastPartitioning
    //    将输入的数据复制到不同的下游算子的并行Tasks实例中，下游算子中的Tasks可以直接从本地内存中获取广播数据集，不再依赖于网络传输
    //    适合于小数据集，当大数据及关联到小数据集的时候可以通过该方式
    val broadcastStream: DataStream[(String, Int)] = dataStream.broadcast

    // 自定义分区customPartitioner
    // partitionCustom应用分区器，第二个参数指定分区器使用到的字段
    val customStream = dataStream.partitionCustom(customPartitioner, 1)
    dataStream.partitionCustom(customPartitioner, "field_name")

    //输出计算结果
    customStream.print()

    env.execute("Streaming PartitioningTransformation")
  }

  object customPartitioner extends Partitioner[String] {
    //获取随机数生成器
    val r = scala.util.Random

    override def partition(key: String, numPartitions: Int): Int = {
      //定义分区策略，key中如果包含a则放在0分区中，其他情况则根据Partitions
      if (key.contains("a")) 0 else r.nextInt(numPartitions)
    }
  }

}
